package com.atguigu.gmall.realtime.util;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.annotation.NoSink;
import com.atguigu.gmall.realtime.bean.TableProcess;
import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.sink.PhoenixSink;
import lombok.SneakyThrows;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.lang.reflect.Field;
import java.nio.charset.StandardCharsets;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Properties;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/8/24 12:07
 */
public class FlinkSinkUtil {
    public static FlinkKafkaProducer<String> getKafkaSink(String topic) {
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "hadoop162:9092,hadoop163:9092,hadoop164:9092");
        props.setProperty("transaction.timeout.ms", 15 * 60 * 1000 + "");
        
        return new FlinkKafkaProducer<String>(
            "default",
            new KafkaSerializationSchema<String>() {
                @Override
                public ProducerRecord<byte[], byte[]> serialize(String value,
                                                                @Nullable Long timestamp) {
                    return new ProducerRecord<>(topic, value.getBytes(StandardCharsets.UTF_8));
                }
            },
            props,
            FlinkKafkaProducer.Semantic.EXACTLY_ONCE
        );
    }
    
    public static SinkFunction<Tuple2<JSONObject, TableProcess>> getKafkaSink() {
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "hadoop162:9092,hadoop163:9092,hadoop164:9092");
        props.setProperty("transaction.timeout.ms", 15 * 60 * 1000 + "");
        
        return new FlinkKafkaProducer<Tuple2<JSONObject, TableProcess>>(
            "default",
            new KafkaSerializationSchema<Tuple2<JSONObject, TableProcess>>() {
                @Override
                public ProducerRecord<byte[], byte[]> serialize(Tuple2<JSONObject, TableProcess> element,
                                                                @Nullable Long timestamp) {
                    String topic = element.f1.getSink_table();
                    byte[] value = element.f0.toJSONString().getBytes(StandardCharsets.UTF_8);
                    return new ProducerRecord<>(topic, value);
                }
            },
            props,
            FlinkKafkaProducer.Semantic.EXACTLY_ONCE
        );
    }
    
    public static SinkFunction<Tuple2<JSONObject, TableProcess>> getHbaseSink() {
        
        return new PhoenixSink();
    }
    
    public static <T> SinkFunction<T> getClickhouseSink(String ckDb,
                                                        String ckTable,
                                                        Class<T> tClass) {
        // 使用jdbcsink 来封装 clickhouse
        
        String driver = Constant.CLICKHOSUE_DRIVER;
        String url = Constant.CLICKHOSUE_URL_PRE + ckDb;
        
        // insert into t(stt, edt, vc,...)values(?, ?, ?,...)
        StringBuilder sql = new StringBuilder();
        sql.append("insert into ").append(ckTable).append("(");
        // TODO 拼接字段
        Field[] fields = tClass.getDeclaredFields();
        String cs = "";
        for (int i = 0; i < fields.length; i++) {
            
            Field field = fields[i];
            NoSink noSink = field.getAnnotation(NoSink.class); // 获取注解: 如果是null表示没有这个注解
            if (noSink == null) {
                if (i < fields.length - 1) {
                    cs += fields[i].getName() + ",";
                } else {
                    cs += fields[i].getName();
                }
            }
            
        }
        cs = cs.replaceAll(",$", ""); // 有了noSink之后, 如果最后一个字段加上NoSink, sql会多一个逗号, 去掉逗号
        sql.append(cs).append(")values(");
        // TODO 拼接占位符
        sql.append(cs.replaceAll("[^,]+", "?")).append(")");
        
        return getJDBCSink(sql.toString(), driver, url);
    }
    
    public static <T> SinkFunction<T> getJDBCSink(String sql,
                                                  String driver,
                                                  String url) {
        return JdbcSink.sink(
            sql,
            new JdbcStatementBuilder<T>() {
                @SneakyThrows
                @Override
                public void accept(PreparedStatement ps,  // sql中有占位符,
                                   T t) throws SQLException {
                    // TODO 给站位符号赋值
                    Class<?> tClass = t.getClass();
                    Field[] fields = tClass.getDeclaredFields();
                    
                    for (int i = 0, position = 1; i < fields.length; i++) {
                        Field field = fields[i];
                        NoSink noSink = field.getAnnotation(NoSink.class);
                        if (noSink == null) {
                            field.setAccessible(true); // 允许访问私有属性
                            Object v = field.get(t);
                            ps.setObject(position++, v);
                        }
                    }
                }
            },
            new JdbcExecutionOptions.Builder()
                .withBatchIntervalMs(2000)  // 默认是0, 表示不会按照时间向外写出
                .withBatchSize(1024)  // 默认是5000
                .withMaxRetries(5)  // 默认是3
                .build(),
            new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                .withDriverName(driver)
                .withUrl(url)
                .build()
        );
    }
}
